/*
 * Copyright (c) 2013, marco.tamburelli@gmail.com
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met: 
 * 
 * 1. Redistributions of source code must retain the above copyright notice, this
 *    list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution. 
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
 * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package org.omorphdb.core.index.hash;

import java.io.IOException;
import java.util.Date;
import java.util.Iterator;
import java.util.NoSuchElementException;

import org.omorphdb.core.exception.CorruptedStorageException;
import org.omorphdb.core.exception.WrappedIOException;
import org.omorphdb.core.index.ChainedIndex;
import org.omorphdb.core.index.IndexType;
import org.omorphdb.core.index.structure.Pointer;
import org.omorphdb.core.io.Constants;
import org.omorphdb.core.io.Resource;
import org.omorphdb.core.value.ValueType;

/**
 * Implementation of an hash map on file for indexing.
 * 
 * @author Marco Tamburelli
 * 
 * @param <T>
 */
public class HashIndex<T> implements ChainedIndex<T>
{
	private final Resource resource;
	private final ValueType type;

	private final Statistics statistics;
	private final HashFunctionFactory<T> hashFunctionFactory;
	private HashFunction<T> hashFunction = null;

	private HashReader<?> reader;
	private HashWriter<?> writer;

	/**
	 * Open an index contained in the provided resource, or initialize the
	 * resource making it to be an index.<br>
	 * The default number of initial slots is 256.
	 * 
	 * @param resource
	 * @param type
	 * @throws IOException
	 */
	public HashIndex(Resource resource, HashFunctionFactory<T> hashFunctionFactory) throws IOException
	{
		this.resource = resource;
		this.type = hashFunctionFactory.keyType();

		this.hashFunctionFactory = hashFunctionFactory;

		initReaderWriter();

		/*
		 * If the resource length is too small, it means that the index is
		 * empty; in such case empty slots will be initialized.
		 */
		if (this.resource.getResourceLimit() < Statistics.SIZE)
		{
			this.statistics = this.writer.appendEmptyStatistics();
			this.hashFunction = this.hashFunctionFactory.getHashFunction(this.statistics.algorithm());

			this.writer.prepareSlots(this.hashFunction.maxValue());
		}
		else
		{
			this.statistics = this.reader.readStatistics();
			this.hashFunction = this.hashFunctionFactory.getHashFunction(this.statistics.algorithm());
		}
	}

	@Override
	public IndexType getIndexType()
	{
		return IndexType.HASH;
	}

	@Override
	public Iterator<Long> search(final T key) throws IOException
	{
		return new Iterator<Long>()
		{
			private Pointer p;

			private Long next;
			private boolean hasNext;

			private WrappedIOException exception;

			{
				long idx = hash(key);
				Slot slot = HashIndex.this.reader.readSlot(idx);

				this.p = slot;

				tryNext();
			}

			private void tryNext()
			{
				try
				{
					/*
					 * No pointer could point to the beginning of a hash index
					 * file, since at the beginning there is the statistics
					 * block.
					 */
					while (this.p.target() > 0)
					{
						Bucket<?> bucket = HashIndex.this.reader.readBucket(this.p);

						try
						{
							/*
							 * In case of collisions we have to returns only
							 * those exactly matching the key ...
							 */
							if (bucket.value().equals(key))
							{
								this.next = Long.valueOf(bucket.target());
								this.hasNext = true;

								return;
							}
						}
						finally
						{
							this.p = bucket.next();
						}
					}

					this.next = null;
					this.hasNext = false;
				}
				catch (IOException e)
				{
					this.exception = new WrappedIOException(e);
				}
			}

			@Override
			public boolean hasNext()
			{
				if (this.exception == null) return this.hasNext;
				else throw this.exception;
			}

			@Override
			public Long next()
			{
				if (this.hasNext) try
				{
					return this.next;
				}
				finally
				{
					tryNext();
				}
				else throw new NoSuchElementException();
			}

			@Override
			public void remove()
			{
				throw new UnsupportedOperationException();
			}
		};
	}

	@Override
	public void register(T key, long docPosition) throws IOException
	{
		long idx = hash(key);
		Slot slot = this.reader.readSlot(idx);

		Pointer p = slot;

		boolean noCollisionDetected = true;
		boolean emptyBucket = true;

		/*
		 * No pointer could point to the beginning of a hash index file, since
		 * at the beginning there is the statistics block.
		 */
		while (p.target() > 0)
		{
			Bucket<?> bucket = this.reader.readBucket(p);

			/*
			 * If an existing bucket points to the document we want to register
			 * nothing should happen, since it means that the same document was
			 * already registered.
			 */
			if (bucket.target() == docPosition) return;

			if (this.statistics.COULD_BE_REBUILT && noCollisionDetected && key.equals(bucket.value()))
			{
				noCollisionDetected = false;
				this.statistics.notifyNewCollision();

				if (this.statistics.shouldBeRebuilt())
				{
					// TODO startRebuild();
				}
			}

			p = bucket.next();

			if (emptyBucket) emptyBucket = false;
		}

		if (emptyBucket) this.statistics.notifyFilledSlot();

		Bucket<?> newBucket = this.writer.appendBucket(key, docPosition);
		p.target(newBucket.position());
		this.writer.updatePointer(p);
	}

	@Override
	public void unregister(T key, long docPosition) throws IOException
	{
		long idx = hash(key);
		Slot slot = this.reader.readSlot(idx);

		Pointer p = slot;

		boolean noCollisionDetected = true;

		/*
		 * No pointer could point to the beginning of a hash index file, since
		 * at the beginning there is the statistics block.
		 */
		while (p.target() > 0)
		{
			Bucket<?> bucket = this.reader.readBucket(p);

			/*
			 * Each possible collision of this key encountered before inserting
			 * an element with the same key, should be removed.
			 */
			if (this.statistics.COULD_BE_REBUILT && noCollisionDetected && key.equals(bucket.value()))
			{
				noCollisionDetected = false;
				this.statistics.notifyRemovedCollision();
			}

			/*
			 * It shouldn't be possible that more than one bucket points to the
			 * same document, so once we find the proper document, we know
			 * exactly which bucket to exclude.
			 */
			if (bucket.target() == docPosition)
			{
				long target = bucket.next().target();
				p.target(target);

				if (p == slot && target <= 0) this.statistics.notifySlotReleased();

				this.writer.updatePointer(p);
				this.writer.updateStatistics(this.statistics);
				break;
			}

			p = bucket.next();
		}
	}

	@Override
	public ValueType getKeysType()
	{
		return this.type;
	}

	/**
	 * It's the used hash function: it provides the slot position from a key.
	 * 
	 * @param key
	 * @return
	 */
	protected long hash(T key)
	{
		return this.hashFunction.calculate(key);
	}

	protected Statistics getStatistics()
	{
		return this.statistics;
	}

	private void initReaderWriter() throws IOException
	{
		switch (this.type)
		{
		case BYTE:
			this.reader = new HashReader<Byte>(this.resource)
			{
				@Override
				protected Byte readValue() throws IOException
				{
					return super.reader.readByte();
				}
			};
			this.writer = new HashWriter<Byte>(this.resource)
			{
				@Override
				protected int getValueSize(Byte value)
				{
					return Constants.SIZEOF_BYTE;
				}

				@Override
				protected void appendValue(Byte value) throws IOException
				{
					super.appender.appendByte(value);
				}
			};
			break;

		case CHAR:
			this.reader = new HashReader<Character>(this.resource)
			{
				@Override
				protected Character readValue() throws IOException
				{
					return super.reader.readChar();
				}
			};
			this.writer = new HashWriter<Character>(this.resource)
			{
				@Override
				protected int getValueSize(Character value)
				{
					return Constants.SIZEOF_CHAR;
				}

				@Override
				protected void appendValue(Character value) throws IOException
				{
					super.appender.appendChar(value);
				}
			};
			break;

		case SHORT:
			this.reader = new HashReader<Short>(this.resource)
			{
				@Override
				protected Short readValue() throws IOException
				{
					return super.reader.readShort();
				}
			};
			this.writer = new HashWriter<Short>(this.resource)
			{
				@Override
				protected int getValueSize(Short value)
				{
					return Constants.SIZEOF_SHORT;
				}

				@Override
				protected void appendValue(Short value) throws IOException
				{
					super.appender.appendShort(value);
				}
			};
			break;

		case INT:
			this.reader = new HashReader<Integer>(this.resource)
			{
				@Override
				protected Integer readValue() throws IOException
				{
					return super.reader.readInt();
				}
			};
			this.writer = new HashWriter<Integer>(this.resource)
			{
				@Override
				protected int getValueSize(Integer value)
				{
					return Constants.SIZEOF_INT;
				}

				@Override
				protected void appendValue(Integer value) throws IOException
				{
					super.appender.appendInt(value);
				}
			};
			break;

		case FLOAT:
			this.reader = new HashReader<Float>(this.resource)
			{
				@Override
				protected Float readValue() throws IOException
				{
					return super.reader.readFloat();
				}
			};
			this.writer = new HashWriter<Float>(this.resource)
			{
				@Override
				protected int getValueSize(Float value)
				{
					return Constants.SIZEOF_FLOAT;
				}

				@Override
				protected void appendValue(Float value) throws IOException
				{
					super.appender.appendFloat(value);
				}
			};
			break;

		case LONG:
			this.reader = new HashReader<Long>(this.resource)
			{
				@Override
				protected Long readValue() throws IOException
				{
					return super.reader.readLong();
				}
			};
			this.writer = new HashWriter<Long>(this.resource)
			{
				@Override
				protected int getValueSize(Long value)
				{
					return Constants.SIZEOF_LONG;
				}

				@Override
				protected void appendValue(Long value) throws IOException
				{
					super.appender.appendLong(value);
				}
			};
			break;

		case DATE:
			this.reader = new HashReader<Date>(this.resource)
			{
				@Override
				protected Date readValue() throws IOException
				{
					return new Date(super.reader.readLong());
				}
			};
			this.writer = new HashWriter<Date>(this.resource)
			{
				@Override
				protected int getValueSize(Date value)
				{
					return Constants.SIZEOF_LONG;
				}

				@Override
				protected void appendValue(Date value) throws IOException
				{
					super.appender.appendLong(value.getTime());
				}
			};
			break;

		case DOUBLE:
			this.reader = new HashReader<Double>(this.resource)
			{
				@Override
				protected Double readValue() throws IOException
				{
					return super.reader.readDouble();
				}
			};
			this.writer = new HashWriter<Double>(this.resource)
			{
				@Override
				protected int getValueSize(Double value)
				{
					return Constants.SIZEOF_DOUBLE;
				}

				@Override
				protected void appendValue(Double value) throws IOException
				{
					super.appender.appendDouble(value);
				}
			};
			break;

		case STRING:
			this.reader = new HashReader<String>(this.resource)
			{
				@Override
				protected String readValue() throws IOException
				{
					return super.reader.readString();
				}
			};
			this.writer = new HashWriter<String>(this.resource)
			{
				@Override
				protected void appendValue(String value) throws IOException
				{
					super.appender.appendString(value);
				}

				@Override
				protected int getValueSize(String value)
				{
					return Constants.SIZEOF_INT + value.getBytes().length;
				}
			};
			break;

		default:
			/*
			 * By default it doesn't do anything; null pointer exception is then
			 * expected when some operation is requested.
			 * 
			 * Note that in any case booleans, arrays and document shouldn't be
			 * used to define a hash index, so CorruptedStorageException will be
			 * thrown here.
			 */
			throw new CorruptedStorageException(this.type.toString()
					+ " shouldn't be used as key type to build a hash index.");
		}
	}
}
